package org.yuanzheng.sink

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
 * @author yuanzheng
 * @date 2020/6/16-13:07
 */
object RedisDBSink {
  def main(args: Array[String]): Unit = {
    // 1.配置环境
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    // 2.读取数据
    val stream: DataStream[String] = streamEnv.socketTextStream("192.168.12.118", 8888)
    // 3.计算
    val result: DataStream[(String, Int)] = stream.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1)
    // 4.结果写入redis
    // redis连接配置
    val jedisPoolConfig = new FlinkJedisPoolConfig.Builder().setDatabase(0).setHost("127.0.0.1").setPort(6379).build()
    // redis的Sink配置
    result.addSink(new RedisSink[(String, Int)](jedisPoolConfig, new RedisMapper[(String, Int)] {
      // 设置redis命令
      override def getCommandDescription: RedisCommandDescription = {
        new RedisCommandDescription(RedisCommand.HSET, "t_wc")
      }

      override def getKeyFromData(t: (String, Int)): String = {
        t._1
      }

      override def getValueFromData(t: (String, Int)): String = {
        t._2 + ""
      }
    }))
    streamEnv.execute();
  }

}
